package cc.chinagps.gboss.mq.reader;

import cc.chinagps.gboss.common.Config;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.jms.*;

@Data
@Slf4j
public abstract class AbstractReader extends Thread implements IReader {
    protected String readerName;
    protected int type;
    private Session session;
    private MessageConsumer consumer;

    public void init() throws Exception {
        Destination destination;
        if (type == 1) {
            destination = session.createQueue(readerName);
        } else {
            destination = session.createTopic(readerName);
        }
        consumer = session.createConsumer(destination);
    }

    private int commitSession() throws JMSException {
        if (batchCount > 0) {
            int temp = batchCount;
            session.commit();
            batchCount = 0;
            return temp;
        }
        return 0;
    }

    private int batchCount = 0;

    @Override
    public void startService() {
        try {
            init();
            startRead();
            start();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }

    public void startRead() {
        try {
            consumer.setMessageListener((msg) -> {
                lastReadTime = System.currentTimeMillis();
                try {
                    if (BytesMessage.class.isInstance(msg)) {
                        BytesMessage byteMsg = (BytesMessage) msg;
                        int len = (int) byteMsg.getBodyLength();
                        byte[] data = new byte[len];
                        byteMsg.readBytes(data);
                        handleMQData(data);
                    } else {
                        log.error("msg type error");
                    }
                    synchronized (session) {
                        batchCount++;
                        if (batchCount >= Config.MQ.GPS_READ_BATCH_SIZE) {
                            commitSession();
                        }
                    }
                } catch (Throwable e) {
                    log.error(e.getMessage(), e);
                }
            });
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private long lastReadTime;
    private long lastCommitTime;

    @Override
    public void run() {
        while (true) {
            try {
                sleep(Config.MQ.GPS_READ_COMMIT_INTERVAL);
                long now = System.currentTimeMillis();
                long temp = lastReadTime;
                if (temp == lastCommitTime || (now - lastReadTime) < Config.MQ.GPS_READ_COMMIT_INTERVAL) {
                    continue;
                }
                synchronized (session) {
                    lastCommitTime = temp;
                    commitSession();
                }
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    public abstract void handleMQData(byte[] data);
}